package edu.upf.opencities.dao.connection;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import me.prettyprint.cassandra.model.CqlQuery;
import me.prettyprint.cassandra.model.CqlRows;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.FailoverPolicy;
import me.prettyprint.cassandra.service.OperationType;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.ConsistencyLevelPolicy;
import me.prettyprint.hector.api.HConsistencyLevel;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.HSuperColumn;
import me.prettyprint.hector.api.beans.OrderedRows;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.beans.Rows;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.MutationResult;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.ColumnQuery;
import me.prettyprint.hector.api.query.MultigetSliceQuery;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.RangeSlicesQuery;
import me.prettyprint.hector.api.query.SuperColumnQuery;

import org.apache.cassandra.auth.SimpleAuthenticator;
import org.apache.cassandra.thrift.AuthenticationException;
import org.apache.cassandra.thrift.AuthenticationRequest;
import org.apache.cassandra.thrift.AuthorizationException;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

import edu.upf.opencities.datamodel.GlobalVar;
import edu.upf.opencities.datamodel.User;
 


public class CassandraUserConnectorImpl implements CassandraUserConnector {

	public static Logger logger = Logger.getLogger(CassandraUserConnectorImpl.class);
	
//	private TTransport tr = null;
//	private static Cassandra.Client client = null;
	private static final StringSerializer se = new StringSerializer();
	private Cluster cluster = null;
	private Keyspace ks = null; 
	
	
    private Properties props;
	private String cassandraHOST;
	private String cassandraPORT;
	private String cassandraCluster;
	private String cassandraKeySpace;
    private String cassandraUserName;
	private String cassandraPassword;
	
	public CassandraUserConnectorImpl(){
        //Properties management
        props = new Properties();

        try {
        	//props.load(new FileInputStream("cassandra-conf.properties") );
        	//props.load(this.getClass().
            //	getResourceAsStream("/edu/upf/opencities/dao/connection/cassandra-conf.properties"));
        	props.load(this.getClass().
            	getResourceAsStream("/cassandra-conf.properties"));
            this.cassandraHOST = props.getProperty("cassandraHOST");
            this.cassandraPORT = props.getProperty("cassandraPORT");
            this.cassandraCluster = props.getProperty("cassandraCluster");
            this.cassandraKeySpace = props.getProperty("cassandraKeySpace");
        	this.cassandraUserName = props.getProperty("cassandraUserName");
            this.cassandraPassword = props.getProperty("cassandraPassword");
    		
        } catch (IOException e) {
            e.printStackTrace();
        }
	}

	public void connect() {
		// Create a cluster
		this.cluster = HFactory.getOrCreateCluster(cassandraCluster, cassandraHOST+":"+cassandraPORT);
		// Choose a keyspace with authentication
		Map<String, String> credentials = new HashMap<String, String>();
		credentials.put(SimpleAuthenticator.USERNAME_KEY, cassandraUserName);
		credentials.put(SimpleAuthenticator.PASSWORD_KEY, cassandraPassword);
		final ConsistencyLevelPolicy policy = new ConsistencyLevelPolicy() {
			@Override
			public HConsistencyLevel get(OperationType op) {
				// TODO Auto-generated method stub
				return HConsistencyLevel.ONE;
			}
			@Override
			public HConsistencyLevel get(OperationType op, String cfName) {
				// TODO Auto-generated method stub
				return HConsistencyLevel.ONE;
			}
		};

		this.ks = HFactory.createKeyspace(cassandraKeySpace, cluster, policy, FailoverPolicy.FAIL_FAST, credentials);
		logger.info("Connected to Cluster: "+cassandraCluster+ "in KeySpace: " + cassandraKeySpace);
	}
	
	public void disconnect() {
	    ks = null;
	    cluster = null;
	    logger.info("Disconnected from Cluster: "+cassandraCluster+ " in KeySpace: " + cassandraKeySpace);
	}
	
	
	//Old connection method
	/*
	public void connect() {

		this.tr = new TFramedTransport(
				new TSocket(cassandraHOST, Integer.parseInt(cassandraPORT)));
		this.client = new Cassandra.Client(new TBinaryProtocol(tr));
		Map<String, String> credentials = new HashMap<String, String>();
		credentials.put(SimpleAuthenticator.USERNAME_KEY, userName);
		credentials.put(SimpleAuthenticator.PASSWORD_KEY, password);
		try {
			tr.open();
			client.login(new AuthenticationRequest(credentials));
			System.out.println("Connected to " + cassandraHOST + ":" + cassandraPORT + ".");

			
		} catch (TTransportException e) {
			System.out
					.println("Error connnecting to " + cassandraHOST + ":" + cassandraPORT + ".");
			e.printStackTrace();
			System.exit(-1);
		} catch (AuthenticationException e) {
			System.out.println("User does not exist or wrong credentials.");
			e.printStackTrace();
			System.exit(-1);
		} catch (AuthorizationException e) {
			System.out.println("Denied access to keyspace " + keySpace + ".");
			e.printStackTrace();
			System.exit(-1);
		} catch (TException e) {
			System.out.println("Error");
			e.printStackTrace();
			System.exit(-1);
		}

	}
*/

	//Old disconnection method
	/*
	public void disconnect() {
		if (tr.isOpen() && tr != null) {
			tr.close();
			System.out.println("Disconnect.");
		} else {
			System.out.println("No connected.");
		}
	}
*/
	//Example method
	public void connectH() {
		
		// Set ColumnFamily name
		String cf = "Users";
		//Set row name
		String rowKey = "manuel.palacin@upf.edu";
		
		// Create Column
		HColumn<String, String> colAux = HFactory.createColumn("mail", "manuel.palacin@upf.edu", se, se);
		
		// Insert column
		Mutator<String> m = HFactory.createMutator(ks, se);
	    MutationResult mr = m.insert(rowKey, cf, colAux);
	    
	    // Get process time 
	    System.out.println("Process time: " + mr.getExecutionTimeMicro());
	    
	    // get value
	    ColumnQuery<String, String, String> q = HFactory.createColumnQuery(ks, se, se, se);
	    q.setName("mail").setColumnFamily(cf);
	    QueryResult<HColumn<String, String>> r = q.setKey(rowKey).execute();
	    
	    HColumn<String, String> col = r.get();

	    String value = col.getValue();
	    System.out.println("Column value: "+value);
	    
	    String name = col.getName();
	    System.out.println("Column name: " +name);
	    
	    //remove value
	    m.delete(rowKey, cf, "mail", se);
	    
	    // test after removal
	    r = q.setKey(rowKey).execute();
	    col = r.get();
		System.out.println("Delete completed, the query must return Null: " + col);
		
	}

	//Example method Super Column
	public void connectHSuper() {

		
		// Set ColumnFamily name
		String cf = "Services";
		//Set row names
		String rowKey1 = "mailUser1"; //In this example we are going to use just one Row
		String rowKey2 = "mailUser2";
		//Set super column names
		String scKey1 = "IDService1"; //In this example we are going to use just one Super Column
		String scKey2 = "IDService2";
		
		// We create a List of columns using a pair of columns 
		@SuppressWarnings("unchecked")
		List<HColumn<String, String>> columns = Arrays.asList(
		    		HFactory.createColumn("serviceName", "Temperature Service", se, se),
		    		HFactory.createColumn("serviceProvider", "UPF", se, se));
		
		//We create a super column
		HSuperColumn superCol = HFactory.createSuperColumn(scKey1, columns, se, se, se);
		// We insert the columns in a super columns inside the database
		Mutator<String> m = HFactory.createMutator(ks, se);		
		m.insert(rowKey1, cf, superCol	);

		// get value
		SuperColumnQuery<String, String, String, String> q = HFactory.createSuperColumnQuery(ks, se, se, se, se);
		q.setSuperName(scKey1).setColumnFamily(cf);
		QueryResult<HSuperColumn<String, String, String>> r = q.setKey(rowKey1).execute();

		HSuperColumn<String, String, String> sc = r.get();

		System.out.println("Number of columns in the super column: " +sc.getSize());
		
		HColumn<String, String> c = sc.get(0);
		String name = c.getName();
		String value = c.getValue();
		System.out.println("Column 1 has name: " + name + " and value: " + value);


		HColumn<String, String> c2 = sc.get(1);
		String name2 = c2.getName();
		String value2 = c2.getValue();
		System.out.println("Column 2 has name: " + name2 + " and value: " + value2);

	
		// remove value
		//m = HFactory.createMutator(ks, se);
		m.subDelete(rowKey1, cf, scKey1, null, se, se);

		// test after removal
		r = q.execute();
		sc = r.get();
		System.out.println("Delete completed, the query must return Null: " +sc);
		
		
	}

	@Override
	public void registerUser(String mail, boolean isAdmin, String userName,
			String userLastName, String userPassword, 
			String company, String country) {
		
		// Set ColumnFamily name
		String cf = "Users";
		//Set row name
		String rowKey = mail;
		
		// Create Columns
		HColumn<String, String> colMail = HFactory.createColumn("mail", mail, se, se);
		HColumn<String, String> colIsAdmin = HFactory.createColumn("isAdmin",new Boolean(isAdmin).toString(), se, se);
		HColumn<String, String> colUserName = HFactory.createColumn("userName", userName, se, se);
		HColumn<String, String> colUserLastName = HFactory.createColumn("userLastName", userLastName, se, se);
		HColumn<String, String> colUserPassword = HFactory.createColumn("userPassword", userPassword, se, se);
		HColumn<String, String> colCompany = HFactory.createColumn("company", company, se, se);
		HColumn<String, String> colCountry = HFactory.createColumn("country", country, se, se);
		
		// Insert columns
		Mutator<String> m = HFactory.createMutator(ks, se);
	    m.insert(rowKey, cf, colMail);
	    m.insert(rowKey, cf, colIsAdmin);
	    m.insert(rowKey, cf, colUserName);
	    m.insert(rowKey, cf, colUserLastName);
	    m.insert(rowKey, cf, colUserPassword);
	    m.insert(rowKey, cf, colCompany);
	    m.insert(rowKey, cf, colCountry);

	    
	    logger.info("Registered User: "+mail);
		
	}

	@Override
	public boolean loginUser(String mail, String userPassword) {


	    
		// Set ColumnFamily name
		String cf = "Users";
		//Set row name
		String rowKey = mail;
		
		// get value of the variable USER_PASSWORD
	    ColumnQuery<String, String, String> q = HFactory.createColumnQuery(ks, se, se, se);
	    q.setName(GlobalVar.USER_PASSWORD).setColumnFamily(cf);
	    QueryResult<HColumn<String, String>> r = q.setKey(rowKey).execute();
	    
	    HColumn<String, String> col = r.get();

	    if (col != null){
		    if( col.getValue().equals(userPassword)){
		    	logger.info("OK: The stored password is: " + col.getValue() + " and the received password is: " + userPassword);
		    	//System.out.println("El password almacenado es: " + col.getValue() + " y el password recibido: " + userPassword ); 
		    	return true;
		    } else {
		    	logger.info("FAIL: The stored password is: " + col.getValue() + " and the received password is: " + userPassword);
		    	//System.out.println("El password almacenado es: " + col.getValue() + " y el password recibido: " + userPassword ); 
		    	return false;
		    }
	    } else {
	    	logger.info("FAIL: User is not in the System");

	    	return false;
	    }
	    
	    
	}

	@Override
	public void deleteUser(String mail, String userPassword) {
 	    
		// Set ColumnFamily name
		String cf = "Users";
		//Set row name
		String rowKey = mail;
		
		// Delete User
		if (loginUser(mail, userPassword)){
			
			Mutator<String> m = HFactory.createMutator(ks, se);
		    //remove the Row... be careful with "range ghosts", maybe the row still alive
		    m.addDeletion(rowKey, cf);
		    m.execute();
		    logger.info("OK: User "+mail+ " removed");
		}else{
			logger.error("FAIL User "+mail+ " was not removed because password is incorrect");
		}
		
		
	}

	@Override
	public User getUser(String mail) {
		
		User user = null;
		
		// Set ColumnFamily name
		String cf = "Users";
		//Set row name
		String rowKey = mail;
		
		
		// get User
		MultigetSliceQuery<String, String, String> multigetSliceQuery =
		    HFactory.createMultigetSliceQuery(ks, se, se, se);
		multigetSliceQuery.setColumnFamily(cf);
		multigetSliceQuery.setKeys(rowKey); //In this case we want just one row_key
		multigetSliceQuery.setRange("", "", false, 20); //Count variable is the maximum number of columns
		QueryResult<Rows<String, String, String>> result = multigetSliceQuery.execute();
		

		//We obtain all the rows (in this case just one) and we iterate it to build a User
		Rows<String, String, String> orderedRows = result.get();
	
		
	    for (Row<String, String, String> r : orderedRows) {
	        	
	    	logger.info("Columns size:" +r.getColumnSlice().getColumns().size()  );
	    	if (! r.getColumnSlice().getColumns().isEmpty()){
	    		if (r.getColumnSlice().getColumnByName(GlobalVar.MAIL).getValue().equals(rowKey)){
			        user = new User();
	    			user.setName(r.getColumnSlice().getColumnByName(GlobalVar.USER_NAME).getValue());
			        user.setLastName(r.getColumnSlice().getColumnByName(GlobalVar.USER_LAST_NAME).getValue());
			        user.setPassword(r.getColumnSlice().getColumnByName(GlobalVar.USER_PASSWORD).getValue());
			        user.setEmail(r.getColumnSlice().getColumnByName(GlobalVar.MAIL).getValue());
			        user.setCountry(r.getColumnSlice().getColumnByName(GlobalVar.COUNTRY).getValue());
			        user.setCompany(r.getColumnSlice().getColumnByName(GlobalVar.COMPANY).getValue());
			        user.setIsAdmin(Boolean.parseBoolean(r.getColumnSlice().getColumnByName(GlobalVar.IS_ADMIN).getValue()));
		    	}
	    	}
	    }
		
	    if (user==null) {
	        logger.info("User not found... ");  
	    }
	    else {
	    	logger.info("Returning loaded User... ");
	    }
		return user;

	}
	
//	public User getUser(String mail) {
//		
//		User user = null;
//		// Set ColumnFamily name
//		String cf = "Users";
//		//Set row name
//		String rowKey = mail;
//
//		CqlQuery<String, String, String> cqlQuery =
//			   new CqlQuery<String, String, String>(ks,se,se,se);
//		cqlQuery.setQuery("SELECT * FROM "+cf+" WHERE KEY = '"+mail+"'");
//
//		QueryResult<CqlRows<String, String, String>> result = cqlQuery.execute();
//
//		CqlRows<String, String, String> rows = result.get();
//		
//		for (Row<String, String, String> row : rows.getList()) {
//			
//			logger.info("Columns size:" +row.getColumnSlice().getColumns().size()  );
//	    	if (! row.getColumnSlice().getColumns().isEmpty()){
//	    		if (row.getColumnSlice().getColumnByName(GlobalVar.MAIL).getValue().equals(rowKey)){
//			        user = new User();
//	    			user.setUserName(row.getColumnSlice().getColumnByName(GlobalVar.USER_NAME).getValue());
//			        user.setUserLastName(row.getColumnSlice().getColumnByName(GlobalVar.USER_LAST_NAME).getValue());
//			        user.setUserPassword(row.getColumnSlice().getColumnByName(GlobalVar.USER_PASSWORD).getValue());
//			        user.setMail(row.getColumnSlice().getColumnByName(GlobalVar.MAIL).getValue());
//			        user.setCountry(row.getColumnSlice().getColumnByName(GlobalVar.COUNTRY).getValue());
//			        user.setCompany(row.getColumnSlice().getColumnByName(GlobalVar.COMPANY).getValue());
//			        user.setAdmin(Boolean.parseBoolean(row.getColumnSlice().getColumnByName(GlobalVar.IS_ADMIN).getValue()));
//		    	}
//	    	}
//		}
//		
//		return user;
//	}

	@Override
	//Be careful: this method returns only 10 users	
	public List<User> getAllUsers() {
		
		List<User> listUsers = new ArrayList<User>();
		
		// Set ColumnFamily name
		String cf = "Users";
	
		// get Users
        RangeSlicesQuery<String, String, String> rangeSlicesQuery =
            	HFactory.createRangeSlicesQuery(ks, se, se, se);
        rangeSlicesQuery.setColumnFamily(cf);
        rangeSlicesQuery.setKeys("", "");
        rangeSlicesQuery.setRange("", "", false, 20);
        
        rangeSlicesQuery.setRowCount(11); //The number of ROWS-1 that returns this query
        QueryResult<OrderedRows<String, String, String>> result = rangeSlicesQuery.execute();
        OrderedRows<String, String, String> orderedRows = result.get();
		
		
		//We obtain all the rows and we iterate them to build a UserList
		User user = null;
        for (Row<String, String, String> r : orderedRows) {
        	user = new User();
        	user.setName(r.getColumnSlice().getColumnByName(GlobalVar.USER_NAME).getValue());
        	user.setLastName(r.getColumnSlice().getColumnByName(GlobalVar.USER_LAST_NAME).getValue());
        	user.setPassword(r.getColumnSlice().getColumnByName(GlobalVar.USER_PASSWORD).getValue());
        	user.setEmail(r.getColumnSlice().getColumnByName(GlobalVar.MAIL).getValue());
        	user.setCountry(r.getColumnSlice().getColumnByName(GlobalVar.COUNTRY).getValue());
        	user.setCompany(r.getColumnSlice().getColumnByName(GlobalVar.COMPANY).getValue());
        	user.setIsAdmin(Boolean.parseBoolean(r.getColumnSlice().getColumnByName(GlobalVar.IS_ADMIN).getValue()));
        	listUsers.add(user);
        }
		
		return listUsers;
	}
}
